package com.chinamobile.iot.mqtt.spi.impl;

import com.alibaba.fastjson.JSON;
import com.chinamobile.iot.mqtt.schedule.DelayTask;
import com.chinamobile.iot.mqtt.spi.model.Qos1Message;
import com.chinamobile.iot.mqtt.spi.model.WaitAckMessage;
import io.netty.channel.Channel;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.data.redis.core.ListOperations;
import org.springframework.data.redis.core.StringRedisTemplate;

import java.util.concurrent.TimeUnit;

import static com.chinamobile.iot.handler.redis.RedisConst.REDIS_4_KEY_PREFIX;


/**
 * Created by szl on 2017/7/24.
 */
public class SubMessageSaveCallback implements DelayTask.TimerCallback<WaitAckMessage> {
    private Logger logger = LoggerFactory.getLogger(SubMessageSaveCallback.class);

    private ProtocolProcessor protocolProcessor;

    private StringRedisTemplate redisTemplate;

    private final boolean saveToRedis;


    public SubMessageSaveCallback(ProtocolProcessor protocolProcessor, StringRedisTemplate redisTemplate, boolean saveToRedis) {
        this.protocolProcessor = protocolProcessor;
        this.redisTemplate = redisTemplate;
        this.saveToRedis = saveToRedis;
    }


    @Override
    public void onTimer(long time, WaitAckMessage waitAckMessage, DelayTask<WaitAckMessage> delayTask) {
        // 超时的消息放入redis中存储最长3天，最多保留500条消息
        Qos1Message qos1Message = waitAckMessage.getMessage();
        String clientId = qos1Message.getToClientId();
        int round = qos1Message.getRound();
        Channel channel = protocolProcessor.getChannel(clientId);
        if (channel == null) {
            // 设备已经下线
            if (!waitAckMessage.isCleanSession()) {
                saveToRedis(clientId, qos1Message);
            }
        } else {
            // 设备还在线
            if (waitAckMessage.isCleanSession()) {
                // 超过两次还没响应，直接让设备下线
                if (round >= 2) {
                    logger.info("the client dons't ack the publish message 2 round, begin disconnect the client. clientId = {}", clientId);
                    protocolProcessor.offlineClientId(clientId);
                } else {
                    // 再次给设备发送
                    protocolProcessor.route2ClientId(clientId, qos1Message.getData(), round);
                }
            } else {
                if (round >= 2) {
                    logger.info("the client dons't ack the publish message 2 round, begin disconnect the client. clientId = {}", clientId);
                    saveToRedis(clientId, qos1Message);
                    // 强行让设备下线
                    protocolProcessor.offlineClientId(clientId);
                } else {
                    protocolProcessor.route2ClientId(clientId, qos1Message.getData(), round);
                }
            }
        }
    }

    private void saveToRedis(String clientId, Qos1Message qos1Message) {
        if (saveToRedis) {
            ListOperations<String, String> listOperations = redisTemplate.opsForList();
            String redis4Key = REDIS_4_KEY_PREFIX + clientId;
            Long size = listOperations.size(redis4Key);
            if (size == 0) {
                listOperations.rightPush(redis4Key, JSON.toJSONString(qos1Message));
                redisTemplate.expire(clientId, 3, TimeUnit.DAYS);
            } else if (size > 0 && size < 500) {
                listOperations.rightPush(redis4Key, JSON.toJSONString(qos1Message));
            }
        }
    }
}
